package com.spark.lola.consumer;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.spark.lola.entity.warehouse.*;
import com.spark.lola.mapper.StoreSupplierMapper;
import com.spark.lola.service.*;
import com.spark.lola.support.common.constant.CommonConstant;
import com.spark.lola.support.common.util.TimeUtil;
import com.spark.lola.support.warehouse.configuration.WarehouseKafkaTopic;
import com.spark.lola.support.warehouse.constant.WarehouseCache;
import com.spark.lola.support.warehouse.constant.WarehouseRedisPrefix;
import com.spark.lola.dto.warehouse.*;
import com.spark.lola.support.warehouse.service.FeiZhiTongServiceImpl;
import com.spark.lola.support.wms.cache.CollectionCache;
import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.transaction.support.TransactionSynchronizationAdapter;
import org.springframework.transaction.support.TransactionSynchronizationManager;
import org.springframework.util.CollectionUtils;

import java.util.*;

/**
 * @Author: DengYeJun
 * @Date: 2020/6/10 20:19
 */
@Component
public class SyncConsumer {
    private final Logger logger = LoggerFactory.getLogger(getClass());
    private IStoreInfoService iStoreInfoService;
    @Autowired
    private IProductInfoService productInfoService;
    @Autowired
    private IStoreProductService iStoreProductService;
    @Autowired
    private FeiZhiTongServiceImpl feiZhiTongService;
    @Autowired
    private StoreSupplierMapper supplierMapper;
    @Autowired
    private IStoreSupplierService iStoreSupplierService;
    @Autowired
    private IGatherLogService iGatherLogService;
    @Autowired
    private IOrderInfoService orderInfoService;

    @Autowired
    private KafkaTemplate kafkaTemplate;
    @Autowired
    private RedisTemplate<String,String> systemRedis;
    @Autowired
    public SyncConsumer(IStoreInfoService iStoreInfoService) {
        this.iStoreInfoService = iStoreInfoService;
        List<StoreInfo> storeInfos = iStoreInfoService.list();
        logger.info("===载入仓库【{}】密匙===START===", storeInfos.size());
        storeInfos.forEach(o -> {
            if (!StringUtils.isEmpty(o.getRemark())) {
                WarehouseCache.organSecretKeys.put(o.getNo(), o.getRemark());
            }
        });
        logger.info("===载入仓库【{}】密匙===END===", storeInfos.size());
    }
    /**
     * 初始化仓库
     *
     * @param consumerRecord
     */
    @KafkaListener(topics = WarehouseKafkaTopic.INIT_SOTRE)
    @Transactional(rollbackFor = Exception.class)
    public void initStore(ConsumerRecord<String, String> consumerRecord) {
        logger.info("===consumer【INIT_SOTRE】仓库【{}】参数【{}】===START===", consumerRecord.key(), consumerRecord.value());
        StoreInfo storeInfo = JSON.parseObject(consumerRecord.value().toString(), StoreInfo.class);
        // 获取品种类目
        QueryWrapper<ProductInfo> query = new QueryWrapper<>();
        query.eq("used", CommonConstant.Y);
        List<ProductInfo> products = productInfoService.list(query);
        if (CollectionUtils.isEmpty(products)) {
            logger.warn("无货品类目信息!");
            return;
        }
        // 获取仓库品种
        List<StoreProduct> storeProducts = iStoreProductService.getListByStoreId(storeInfo.getId());
        // 遍历品种类目 新增仓库货品记录
        List<StoreProduct> storeProductDetails = new ArrayList<>();
        products.forEach(product -> {
            long count = storeProducts.stream().filter(o -> o.getNo().equals(product.getNo())).count();
            if (count == 0) {
                // 仓库不存在该商品 新增品种
                storeProductDetails.add(new StoreProduct(storeInfo.getId(), product));
            }
        });
        if (storeProductDetails.size() > 0) {
            if (!iStoreProductService.saveBatch(storeProductDetails, 100)) {
                logger.error("仓库【{}】品种添加失败", storeInfo.getId());
            }
        }
        // 发送仓库 供应商同步指令
        TransactionSynchronizationManager.registerSynchronization(
                new TransactionSynchronizationAdapter() {
                    @Override
                    public void afterCommit() {
                        kafkaTemplate.send(WarehouseKafkaTopic.SYNC_STORE_SUPPLIER, storeInfo.getNo(), JSON.toJSONString(storeInfo));
                    }
                });
        logger.info("===consumer【INIT_SOTRE】仓库【{}】参数【{}】===END===", consumerRecord.key(), consumerRecord.value());
    }

    /**
     * 同步品种
     *
     * @param consumerRecord
     */
    @KafkaListener(topics = WarehouseKafkaTopic.SYNC_PRODUCT)
    @Transactional(rollbackFor = Exception.class)
    public void syncProduct(ConsumerRecord<String, String> consumerRecord) {
        logger.info("===consumer【SYNC_PRODUCT】同步品种类目===START===");
        List<ProductInfo> products = productInfoService.list();
        if (CollectionUtils.isEmpty(products)) {
            products = new ArrayList<>();
        }
        List<FztProductInfo> fztProductInfos = feiZhiTongService.getProducts();
        if (CollectionUtils.isEmpty(fztProductInfos)) {
            return;
        }
        List<ProductInfo> details = new ArrayList<>();
        for (FztProductInfo fztProductInfo : fztProductInfos) {
            long count = products.stream().filter(o -> o.getNo().equals(fztProductInfo.getCode())).count();
            if (count == 0) {
                details.add(new ProductInfo(fztProductInfo.getCode(), fztProductInfo.getName()));
            }
        }
        if (details.size() > 0) {
            if (!productInfoService.saveBatch(details, 100)) {
                logger.error("品种类目添加失败");
                return;
            }
            TransactionSynchronizationManager.registerSynchronization(
                    new TransactionSynchronizationAdapter() {
                        @Override
                        public void afterCommit() {
                            details.forEach(detail -> {
                                kafkaTemplate.send(WarehouseKafkaTopic.DEAL_NEW_PRODUCT, detail.getId().toString(), JSON.toJSONString(detail));
                            });
                        }
                    });
        }
        logger.info("===consumer【SYNC_PRODUCT】同步品种类目===END===");
    }

    /**
     * 处理仓库品种
     * @param consumerRecord
     */
    @KafkaListener(topics = WarehouseKafkaTopic.DEAL_NEW_PRODUCT)
    @Transactional(rollbackFor = Exception.class)
    public void dealNewProduct(ConsumerRecord<String, String> consumerRecord) {
        logger.info("===consumer【DEAL_NEW_PRODUCT】货品类目【{}】参数【{}】===START===", consumerRecord.key(), consumerRecord.value());
        ProductInfo productInfo = JSON.parseObject(consumerRecord.value(), ProductInfo.class);
        // 获取仓库列表
        List<StoreInfo> stores = iStoreInfoService.getStores();
        if (CollectionUtils.isEmpty(stores)) {
            return;
        }
        // 获取仓库指定品种类目货品
        List<StoreProduct> storeProducts = iStoreProductService.getListByProductId(productInfo.getId());
        Map<Long, String> map = new HashMap<>();
        if (!CollectionUtils.isEmpty(storeProducts)) {
            storeProducts.forEach(s -> map.put(s.getStoreId(), s.getNo()));
        }
        stores.forEach(store -> {
            if (!map.containsKey(store.getId())) {
                // 保存仓库品种
                iStoreProductService.save(new StoreProduct(store.getId(), productInfo));
            }
        });
        logger.info("===consumer【SYNC_PRODUCT】货品类目【{}】参数【{}】===END===", consumerRecord.key(), consumerRecord.value());
    }

    /**
     * 同步供应商
     *
     * @param consumerRecord
     */
    @KafkaListener(topics = WarehouseKafkaTopic.SYNC_SUPPLIER)
    @Transactional(rollbackFor = Exception.class)
    public void syncSupplier(ConsumerRecord<String, String> consumerRecord) {
        logger.info("===consumer【SYNC_SUPPLIER】同步供应商===START===");
        // 获取仓库列表
        List<StoreInfo> stores = iStoreInfoService.getStores();
        stores.forEach(storeInfo -> {
            kafkaTemplate.send(WarehouseKafkaTopic.SYNC_STORE_SUPPLIER, storeInfo.getNo(), JSON.toJSONString(storeInfo));
        });
        logger.info("===consumer【SYNC_SUPPLIER】同步供应商===END===");
    }

    /**
     * 同步仓库供应商
     * @param consumerRecord
     */
    @KafkaListener(topics = WarehouseKafkaTopic.SYNC_STORE_SUPPLIER)
    @Transactional(rollbackFor = Exception.class)
    public void syncStoreSupplier(ConsumerRecord<String, String> consumerRecord) {
        logger.info("===consumer【SYNC_STORE_SUPPLIER】同步仓库【{}】供应商===START===", consumerRecord.key());
        List<StoreSupplier> storeSuppliers = new ArrayList<>();

        StoreInfo storeInfo = JSON.parseObject(consumerRecord.value(), StoreInfo.class);
        // 获取指定仓库 供应商记录
        List<FztSupplierInfo> lst = feiZhiTongService.getSuppliers(storeInfo.getNo(), null);
        logger.info("获取供应商数量【{}】", lst.size());
        if (CollectionUtils.isEmpty(lst)) {
            return;
        }
        // 获取仓库 商户键值对
        List<String> supplierNos = getSupplierMap(storeInfo.getId());
        for (FztSupplierInfo fztSupplier : lst) {
            if (!supplierNos.contains(fztSupplier.getIcno())) {
                storeSuppliers.add(
                        new StoreSupplier(storeInfo.getId(), fztSupplier.getIcno(), fztSupplier.getUserName())
                );
            }
        }
        if (!CollectionUtils.isEmpty(storeSuppliers)) {
            iStoreSupplierService.saveBatch(storeSuppliers, 10000);
        }
        logger.info("===consumer【SYNC_STORE_SUPPLIER】同步仓库【{}】供应商===END===", consumerRecord.key());
    }

    @KafkaListener(topics = WarehouseKafkaTopic.GATHER_LOG_SAVE)
    @Transactional(rollbackFor = Exception.class)
    public void saveGatherLog(ConsumerRecord<String, String> consumerRecord) {
        GatherLog model = JSON.parseObject(consumerRecord.value(),GatherLog.class);
        model.setLastTime(new Date());
        iGatherLogService.save(model);
    }

    /**
     * 同步仓库订单
     * @param consumerRecord
     */
    @KafkaListener(topics = WarehouseKafkaTopic.SYNC_STORE_ORDER)
    @Transactional(rollbackFor = Exception.class)
    public void syncStoreOrder(ConsumerRecord<String, String> consumerRecord) throws Exception {
        logger.info("===consumer【SYNC_STORE_ORDER】同步仓库【{}】订单=START===", consumerRecord.key());
        JSONObject request = JSON.parseObject(consumerRecord.value());
        String key = WarehouseRedisPrefix.ORGAN_LAST_NO + consumerRecord.key();
        // 获取指定站点最后订单号
        String lastNo = "";
//        if (systemRedis.hasKey(key)) {
//            lastNo = systemRedis.opsForValue().get(key);
//        }
        List<FztPurOrder> lst = new ArrayList<>();
        // 获取订单数据
        try {
            lastNo = bindOrganOrder(lst, request.getString("start"), request.getString("end"), consumerRecord.key(), lastNo);
        }
        catch (Exception e){
            logger.error("{}",e.getMessage(),e);
            throw new Exception(e.getMessage());
        }
        if (CollectionUtils.isEmpty(lst)) {
            logger.error("无符合条件订单数据");
            return;
        }
        // 获取交易日范围本地订单列表
        List<String> localOrderKeys = orderInfoService.getKeysByTradeDay(request.getString("start"), request.getString("end"));
        // 处理存储列表
        List<OrderInfo> orderInfos = new ArrayList<>();
        lst.forEach(o -> {
            String orderKey = o.getOrganCode() + ":" + o.getBno();
            if (!localOrderKeys.contains(orderKey)) {
                // 本地不存在加入存储列表
                orderInfos.add(new OrderInfo(o));
            }
        });
        if (orderInfos.size() == 0) {
            logger.error("无符合条件新增订单数据");
            return;
        }
        if (!orderInfoService.saveBatch(orderInfos,orderInfos.size())) {
            throw new Exception("订单保存失败！");
        }
        for(OrderInfo o:orderInfos) {
            // 发送统计结算通知
            kafkaTemplate.send(WarehouseKafkaTopic.CAL_STORE, o.getStoreId().toString(), JSON.toJSONString(o));
            kafkaTemplate.send(WarehouseKafkaTopic.CAL_PRODUCT, o.getProductNo(), JSON.toJSONString(o));
            kafkaTemplate.send(WarehouseKafkaTopic.CAL_SUPPLIER, o.getSupplierNo(), JSON.toJSONString(o));
        }
        systemRedis.opsForValue().set(key, lastNo);
        logger.info("===consumer【SYNC_STORE_ORDER】同步仓库【{}】订单===END===", consumerRecord.key());
    }

    /**
     * 获取仓库 商户键值对
     * @return
     */
    private List<String> getSupplierMap(Long storeId) {
        List<String> supplierNos = new ArrayList<>();
        QueryWrapper<StoreSupplier> query = new QueryWrapper<>();
        query.eq("store_id", storeId);
        query.eq("used", CommonConstant.Y);
        List<StoreSupplier> suppliers = supplierMapper.selectList(query);
        suppliers.forEach(o -> supplierNos.add(o.getNo()));
        return supplierNos;
    }

    private String bindOrganOrder(List<FztPurOrder> lst, String start, String end, String organCode, String lastOrderNo){
        PageOrder<FztPurOrder> page = feiZhiTongService.getPurOrders(start,end,organCode,null,null,lastOrderNo);
        if(page == null || page.getTotal()==0) {
            return "";
        }
        lst.addAll(page.getList());
        page.getLastNo();
        if(page.getLeft()>0){
            bindOrganOrder(lst,start,end,organCode,lastOrderNo);
        }
        return page.getLastNo();
    }
}
